package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @Author lzc
 * @Date 2022/5/23 9:10
 */
public class Dwd_05_DwdTradeOrderPreProcessApp extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_05_DwdTradeOrderPreProcessApp().init(2005, 2, "Dwd_05_DwdTradeOrderPreProcessApp");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));
        
        
        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_05_DwdTradeOrderPreProcessApp");
        // 2. 读取base_dic 字典表
        readBaseDic(tEnv);
        // 3. 过滤出oder_detail
        // 实际业务中, 详情表只有insert, 不会更新和删除
        Table orderDetail = tEnv.sqlQuery("select " +
                                        " `data`['id'] id, " +
                                        " `data`['order_id'] order_id, " +
                                        " `data`['sku_id'] sku_id, " +
                                        " `data`['sku_name'] sku_name, " +
                                        " `data`['source_id'] source_id, " +
                                        " `data`['source_type'] source_type, " +
                                        " `data`['sku_num'] sku_num, " +
                                        " cast(cast(`data`['order_price'] as decimal(16,2)) * cast(`data`['sku_num'] as decimal(16,2)) as string)  split_original_amount, " +
                                        " `data`['create_time'] create_time, " +
                                        " `data`['split_total_amount'] split_total_amount, " + //分摊原始总金额 -优惠-活动
                                        " `data`['split_activity_amount'] split_activity_amount, " +
                                        " `data`['split_coupon_amount'] split_coupon_amount, " +
                                        " pt , " +
                                        " ts od_ts " +
                                        "from ods_db " +
                                        "where `table`='order_detail' " +
                                        "and `type`='insert'");
        
        tEnv.createTemporaryView("order_detail", orderDetail);
        // 4. 过滤出 order_info
        Table orderInfo = tEnv.sqlQuery("select" +
                                        " `data`['id'] id, " +
                                        " `data`['user_id'] user_id, " +
                                        " `data`['province_id'] province_id, " +
                                        " `data`['operate_time'] operate_time, " +
                                        " `data`['order_status'] order_status, " +
                                        " `old`, " +
                                        " `type`," +
                                        " ts oi_ts " +
                                        "from ods_db " +
                                        "where `table`='order_info' " +
                                        "and (" +
                                        " `type`='insert' " +
                                        "  or `type`='update'" +
                                        ")");
        tEnv.createTemporaryView("order_info", orderInfo);
        
        // 5. 过滤出 oder_detail_activity
        Table orderDetailActivity = tEnv.sqlQuery("select " +
                                        " `data`['order_detail_id'] order_detail_id, " +
                                        " `data`['activity_id'] activity_id, " +
                                        " `data`['activity_rule_id'] activity_rule_id " +
                                        "from ods_db " +
                                        "where `table`='order_detail_activity' " +
                                        "and `type`='insert'");
        tEnv.createTemporaryView("order_detail_activity",orderDetailActivity);
        // 6. 过滤出 oder_detail_coupon
    
        Table orderDetailCoupon = tEnv.sqlQuery("select " +
                                                     " `data`['order_detail_id'] order_detail_id, " +
                                                     " `data`['coupon_id'] coupon_id, " +
                                                     " `data`['coupon_use_id'] coupon_use_id " +
                                                     "from ods_db " +
                                                     "where `table`='order_detail_coupon' " +
                                                     "and `type`='insert'");
        tEnv.createTemporaryView("order_detail_coupon",orderDetailCoupon);
        // 7. 5张表join
        Table resultTable = tEnv.sqlQuery("select " +
                                        "od.id, " +
                                        "od.order_id, " +
                                        "oi.user_id, " +
                                        "oi.order_status, " +
                                        "od.sku_id, " +
                                        "od.sku_name, " +
                                        "oi.province_id, " +
                                        "act.activity_id, " +
                                        "act.activity_rule_id, " +
                                        "cou.coupon_id, " +
                                        "date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                                        "od.create_time, " +
                                        "date_format(oi.operate_time, 'yyyy-MM-dd') operate_date_id, " +
                                        "oi.operate_time, " +
                                        "od.source_id, " +
                                        "od.source_type, " +
                                        "bd_1.dic_name source_type_name, " +
                                        "od.sku_num, " +
                                        "od.split_original_amount, " +
                                        "od.split_activity_amount, " +
                                        "od.split_coupon_amount, " +
                                        "od.split_total_amount, " +
                                        "oi.`type`, " +
                                        "oi.`old`, " +
                                        "od.od_ts, " +
                                        "oi.oi_ts, " +
                                        "current_row_timestamp() row_op_ts " +
                                        "from order_detail od " +
                                        "join base_dic FOR SYSTEM_TIME AS OF od.pt bd_1 on od.source_type=bd_1.dic_code " +  // TODO lookup join
                                        "join order_info oi on od.order_id=oi.id " +
                                        "left join order_detail_activity act on od.id=act.order_detail_id " +
                                        "left join order_detail_coupon cou on od.id=cou.order_detail_id");
        
        // 8.  结果写出到kafka中 upsert-kafka
        tEnv.executeSql("create table dwd_trade_order_pre_process(" +
                            "id string, " +
                            "order_id string, " +
                            "user_id string, " +
                            "order_status string, " +
                            "sku_id string, " +
                            "sku_name string, " +
                            "province_id string, " +
                            "activity_id string, " +
                            "activity_rule_id string, " +
                            "coupon_id string, " +
                            "date_id string, " +
                            "create_time string, " +
                            "operate_date_id string, " +
                            "operate_time string, " +
                            "source_id string, " +
                            "source_type string, " +
                            "source_type_name string, " +
                            "sku_num string, " +
                            "split_original_amount string, " +
                            "split_activity_amount string, " +
                            "split_coupon_amount string, " +
                            "split_total_amount string, " +
                            "`type` string, " +
                            "`old` map<string,string>, " +
                            "od_ts string, " +
                            "oi_ts string, " +
                            "row_op_ts timestamp_ltz(3), " +
                            "primary key(id) not enforced " +
                            ")" + SqlUtil.getKafkaUpsertDDL(Constant.TOPIC_DWD_TRADE_ORDER_PRE_PROCESS));
        
        resultTable.executeInsert("dwd_trade_order_pre_process");
        
        
        
    }
}
/*
字段是函数生产:
ddl 使用函数
    pt as proctime()
    
dml  查询
    proctime() as pt

 */